package com.xiaoming;

import com.xiaoming.utils.*;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.concurrent.Executors.newFixedThreadPool;

public class RpcConsumer {
    private ExecutorService executor = newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    private static Map<String, UserClientHandler> userClientHandlerMap = new HashMap<>();
    static AtomicInteger atomicInteger = new AtomicInteger(0);

    public RpcConsumer() throws Exception {
        ZKBuilder builder = ZKBuilder.build((Address address, String type) -> {
            System.out.println(String.format("子节点改变，%s 服务器 %s",type,address));
            if ("ADD".equals(type)) {
                addServer(address);
                return;
            }
            if ("DELETE".equals(type)) {
                removeServer(address.getHost()+"-"+address.getPort());
                return;
            }
        }, Integer.MAX_VALUE);
        Set<Map.Entry<String, Address>> entries = builder.getServerMap().entrySet();
        for (Map.Entry<String, Address> entry : entries) {
            addServer(entry.getValue());
        }
    }

    public Object createProxy(final Class<?> serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass},
                (Object proxy, Method method, Object[] args) -> {

                    UserClientHandler userClientHandler = getServer();

                    //组装RpcRequest对象
                    RpcRequest rpcRequest = new RpcRequest();
                    rpcRequest.setRequestId(UUID.randomUUID().toString());
                    rpcRequest.setClassName("com.xiaoming.service.UserServiceImpl");
                    rpcRequest.setMethodName("sayHello");
                    Object[] paraObjArr = {args[0]};
                    rpcRequest.setParameters(paraObjArr);
                    Class[] paraTypeArr = {User.class};
                    rpcRequest.setParameterTypes(paraTypeArr);
                    userClientHandler.setParam(rpcRequest);
                    return executor.submit(userClientHandler).get();
                });
    }

    public void removeServer(String key) {
        for (String host :userClientHandlerMap.keySet()){
            if(host.startsWith(key)){
                userClientHandlerMap.remove(host);
                return;
            }
        }
    }
    public boolean isReady(){
        return userClientHandlerMap.size() > 0;
    }

    private UserClientHandler getServer() {
        if (userClientHandlerMap.size() == 0) {
            throw new RuntimeException("服务器列表为空！");
        }
        UserClientHandler userClientHandler = null;
        int andIncrement = atomicInteger.getAndIncrement();
        int index = andIncrement % userClientHandlerMap.size();
        for (String key : userClientHandlerMap.keySet()) {
            if (key.endsWith("$$" + index)) {
                userClientHandler = userClientHandlerMap.get(key);
            }
        }
        ;
        return userClientHandler;
    }

    private String getKey(Address address) {
        return address.getHost() + "-" + address.getPort() + "$$" + userClientHandlerMap.size();
    }

    //2.初始化netty客户端
    private void addServer(Address address) throws InterruptedException {
        String key = getKey(address);
        for (String host :userClientHandlerMap.keySet()){
            if(key.startsWith(host)){
                return;
            }
        }
        UserClientHandler userClientHandler = new UserClientHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                        pipeline.addLast(userClientHandler);
                    }
                });
        bootstrap.connect(address.getHost(), address.getPort()).sync();
        userClientHandlerMap.put(key, userClientHandler);
    }
}
